package hupu.hnlp.hot_buffer.distbuffer.service;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import javax.persistence.EntityManager;
import javax.persistence.NoResultException;
import javax.persistence.TemporalType;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.annotation.JsonRawValue;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;

import hupu.hnlp.hot_buffer.share.rmi.distbuffer.DistBuffer;

@Service("distBuffer")
public class DistBufferService extends BaseService implements DistBuffer
{
    private final static Logger logger = LoggerFactory
            .getLogger(DistBufferService.class);

    public DistBufferService()
    {
        cid_hashes = new CidHash[CID_HASH_NUM];
        for (int i = 0; i < cid_hashes.length; i++)
        {
            cid_hashes[i] = new CidHash();
        }
    }

    private final String error_response = "{\"errno\":9999, \"errmsg\":\"未知错误\"}";

    /* ========== ThreadLocal ========== */
    class Local
    {
        private byte[] buf = new byte[500 * 1024];

        public byte[] getBuf()
        {
            return buf;
        }

        public void setBuf(byte[] buf)
        {
            this.buf = buf;
        }

    }

    private ThreadLocal<Local> local = new ThreadLocal<Local>()
    {
        @Override
        protected Local initialValue()
        {
            return new Local();
        }
    };
    /* ========== ThreadLocal ========== */

    private volatile boolean can_sync = false;

    /* ========== downgrade ========== */
    private final String DOWNGRADE_CODE = "DEGRADED_SERVICE";

    class DowngradeStatus
    {
        public final static int OFF = 0;
        public final static int ON = 1;
    }

    private volatile boolean is_downgrade = false;
    /* ========== downgrade ========== */

    /* ========== data ========== */
    private final int BLACK_EXPIRE_DAYS = 7;
    private volatile Instant sync_black_time = Instant.now()
            .minus(BLACK_EXPIRE_DAYS, ChronoUnit.DAYS);

    class ItemStatus
    {
        public final static int NORMAL = 0;
        public final static int BLACK = 1;
    }

    // private final int CONTENT_EXPIRE_DAYS = 7;
    private volatile Instant sync_content_time = Instant.ofEpochSecond(0);

    class MtInfoType
    {
        public final static String MULTI = "mt"; // 图文
        public final static String VIDEO = "vt"; // 视频
    }

    private ReadWriteLock data_lock = new ReentrantReadWriteLock();
    // lock
    // 缺省热门推荐列表 fid -> [ xid ]
    // private Map<Integer, Map<String, Integer>> hotlist_default = new
    // HashMap<>();

    // 推荐(按话题)默认列表 topic -> { xid -> score }
    // private Map<Integer, Map<String, Integer>> topic_hots_default = new
    // HashMap<>();

    // 黑名单 xid -> 0
    private Map<String, Integer> blacks = new HashMap<>();

    // 内容 xid -> { content }
    class ContentNode
    {
        private String content;
        private int fid;
        private String type;
        private int topic;

        public int getFid()
        {
            return fid;
        }

        public int getTopic()
        {
            return topic;
        }

        public void setTopic(int topic)
        {
            this.topic = topic;
        }

        public void setFid(int fid)
        {
            this.fid = fid;
        }

        public String getContent()
        {
            return content;
        }

        public void setContent(String content)
        {
            this.content = content;
        }

        public String getType()
        {
            return type;
        }

        public void setType(String type)
        {
            this.type = type;
        }

    }

    private Map<String, ContentNode> contents = new HashMap<>();

    private Map<String, Set<String>> video_tags = new HashMap<>();
    private Map<String, List<VideoNode>> tag_videos = new HashMap<>();
    // lock
    /* ========== data ========== */

    public boolean canSync()
    {
        return can_sync;
    }

    public void openSync()
    {
        can_sync = true;
    }

    public void initStorage() throws Exception
    {
        init_dir(appconfig.storage.path);
        for (int i = 0; i < CID_HASH_NUM; i++)
        {
            init_dir(String.format("%s/%d", appconfig.storage.path, i));
        }
    }

    private void init_dir(String dir) throws Exception
    {
        File d = new File(dir);
        if (d.exists())
        {
            if (!d.isDirectory() || !d.canRead() || !d.canWrite()
                    || !d.canExecute())
            {
                throw new Exception(String.format("dir:%s invalid", dir));
            }
        }
        else
        {
            if (!d.mkdir())
            {
                throw new Exception(String.format("dir:%s create fail", dir));
            }
        }
    }

    private void delay()
    {
        try
        {
            Thread.sleep(new Random().nextInt(10) * 1000);
        }
        catch (InterruptedException e)
        {
        }
    }

    public void syncDowngrade()
    {
        EntityManager hot_em = null;

        try
        {
            hot_em = hotEntityManagerFactory.createEntityManager();

            int value = Integer
                    .valueOf(Optional
                            .ofNullable((String) hot_em
                                    .createNativeQuery(
                                            "select value from hot_system_param"
                                                    + " where code=:code")
                                    .setParameter("code", DOWNGRADE_CODE)
                                    .getSingleResult())
                            .map(String::trim).orElse("0"));
            boolean status = value == DowngradeStatus.ON ? true : false;
            if (status != is_downgrade)
            {
                logger.info("sync downgrade {} -> {}", is_downgrade, status);
                is_downgrade = status;
            }
        }
        catch (NoResultException no_e)
        {
            logger.error("DEGRADED_SERVICE not in hot_system_param, bad");
        }
        catch (Exception e)
        {
            logger.error("sync fail: {}", e);
        }
        finally
        {
            if (hot_em != null)
                hot_em.close();
        }
    }

    // public void syncHotListDefault()
    // {
    // delay();
    //
    // long begin_time = Instant.now().toEpochMilli();
    //
    // SyncHotListResult sync_hotlist_result = sync_hotlist(0, null, null,
    // null, null);
    // Map<Integer, Map<String, Integer>> hotlist_default_update =
    // sync_hotlist_result
    // .getHotlist();
    //
    // logger.info("sync hotlist cost {} ms, num({}) size({}) ",
    // Instant.now().toEpochMilli() - begin_time,
    // hotlist_default_update.entrySet().stream()
    // .collect(Collectors
    // .summingInt(entry -> entry.getValue().size())),
    // sync_hotlist_result.getSize());
    //
    // try
    // {
    // data_lock.writeLock().lock();
    //
    // hotlist_default = hotlist_default_update;
    // }
    // catch (Exception e)
    // {
    // logger.error("lock fail: {}", e);
    // return;
    // }
    // finally
    // {
    // data_lock.writeLock().unlock();
    // }
    // }

    public void syncBlack(boolean is_complete)
    {
        delay();

        if (is_complete)
        {
            sync_black_time = Instant.now().minus(BLACK_EXPIRE_DAYS,
                    ChronoUnit.DAYS);
        }

        Map<String, Integer> blacks_update = new HashMap<>();

        EntityManager hot_em = null;

        try
        {
            hot_em = hotEntityManagerFactory.createEntityManager();

            long begin_time = Instant.now().toEpochMilli();
            Instant now = Instant.now();
            ((List<?>) hot_em
                    .createNativeQuery("select xid, status from hot_item_status"
                            + " where update_at>=:update_at")
                    .setParameter("update_at", Date.from(sync_black_time),
                            TemporalType.TIMESTAMP)
                    .getResultList()).forEach(r -> {
                        Object[] row = (Object[]) r;
                        int col = 0;
                        String xid = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        int status = Optional.ofNullable((Short) row[col++])
                                .map(Short::intValue).orElse(-1);
                        if (xid.isEmpty() || status == -1)
                            return;
                        blacks_update.put(xid, status);
                    });
            sync_black_time = now;
            if (!blacks_update.isEmpty())
            {
                logger.info("sync black cost {} ms, size({})",
                        Instant.now().toEpochMilli() - begin_time,
                        blacks_update.size());
            }
        }
        catch (Exception e)
        {
            logger.error("sync fail: {}", e);
            return;
        }
        finally
        {
            if (hot_em != null)
                hot_em.close();
        }

        if (!blacks_update.isEmpty())
        {
            try
            {
                data_lock.writeLock().lock();

                if (is_complete)
                {
                    blacks = blacks_update.entrySet().stream()
                            .filter(x -> x.getValue() == ItemStatus.BLACK)
                            .collect(Collectors.toMap(Entry::getKey, x -> 0));
                }
                else
                {
                    blacks_update.forEach((xid, status) -> {
                        if (status == ItemStatus.NORMAL)
                            blacks.remove(xid);
                        else if (status == ItemStatus.BLACK)
                            blacks.put(xid, 0);
                    });
                }
            }
            catch (Exception e)
            {
                logger.error("lock fail: {}", e);
                return;
            }
            finally
            {
                data_lock.writeLock().unlock();
            }
        }
    }

    public void syncContent(boolean is_complete)
    {
        delay();

        if (is_complete)
        {
            sync_content_time = Instant.ofEpochSecond(0);
        }

        Map<String, ContentNode> contents_update = new HashMap<>();

        EntityManager hot_em = null;

        try
        {
            hot_em = hotEntityManagerFactory.createEntityManager();

            long begin_time = Instant.now().toEpochMilli();

            Instant now = Instant.now();
            ((List<?>) hot_em.createNativeQuery(
                    "select xid, type, json, create_at from hot_news_info"
                            + " where update_at>=:update_at")
                    .setParameter("update_at",
                            sync_content_time.getEpochSecond())
                    .getResultList()).forEach(r -> {
                        Object[] row = (Object[]) r;
                        int col = 0;
                        String xid = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        String type = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        String content = Optional
                                .ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        long create_at = Optional
                                .ofNullable((Integer) row[col++]).orElse(0)
                                & 0xFFFFFFFFL;
                        if (xid.isEmpty() || content.isEmpty()
                                || create_at <= 0)
                            return;
                        ContentNode node = new ContentNode();
                        // node.setCreate_time(Instant.ofEpochSecond(create_at));
                        node.setContent(content);
                        node.setType(type);
                        contents_update.put(xid, node);
                    });
            ((List<?>) hot_em.createNativeQuery(
                    "select xid, type, fid, json, create_at from hot_mt_info"
                            + " where update_at>=:update_at")
                    .setParameter("update_at",
                            sync_content_time.getEpochSecond())
                    .getResultList()).forEach(r -> {
                        Object[] row = (Object[]) r;
                        int col = 0;
                        String xid = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        String type = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        String fid_s = Optional.ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        String content = Optional
                                .ofNullable((String) row[col++])
                                .map(String::trim).orElse("");
                        long create_at = Optional
                                .ofNullable((Integer) row[col++]).orElse(0)
                                & 0xFFFFFFFFL;
                        if (xid.isEmpty() || fid_s.isEmpty()
                                || content.isEmpty() || create_at <= 0)
                            return;
                        int fid;
                        try
                        {
                            fid = Integer.valueOf(fid_s);
                        }
                        catch (Exception e)
                        {
                            logger.error(
                                    "xid({}) in hot_mt_info: fid({}) invalid",
                                    xid, fid_s);
                            return;
                        }
                        ContentNode node = new ContentNode();
                        // node.setCreate_time(Instant.ofEpochSecond(create_at));
                        node.setContent(content);
                        node.setFid(fid);
                        node.setType(type);
                        contents_update.put(xid, node);
                    });
            sync_content_time = now;

            logger.info(
                    "sync content complete({}) cost {} ms, num({}) size({}) ",
                    is_complete, Instant.now().toEpochMilli() - begin_time,
                    contents_update.size(),
                    contents_update.values().stream()
                            .collect(Collectors.summingLong(x -> (long) x
                                    .getContent().getBytes().length)));
        }
        catch (Exception e)
        {
            logger.error("sync fail: {}", e);
            return;
        }
        finally
        {
            if (hot_em != null)
                hot_em.close();
        }

        if (!contents_update.isEmpty())
        {
            try
            {
                data_lock.writeLock().lock();

                if (is_complete)
                {
                    contents = contents_update;
                }
                else
                {
                    contents.putAll(contents_update);
                }
                logger.info("contents({})", contents.size());
            }
            catch (Exception e)
            {
                logger.error("lock fail: {}", e);
                return;
            }
            finally
            {
                data_lock.writeLock().unlock();
            }
        }
    }

    class VideoNode
    {
        private String xid;
        private String mtag;
        private int score;

        public String getXid()
        {
            return xid;
        }

        public void setXid(String xid)
        {
            this.xid = xid;
        }

        public String getMtag()
        {
            return mtag;
        }

        public void setMtag(String mtag)
        {
            this.mtag = mtag;
        }

        public int getScore()
        {
            return score;
        }

        public void setScore(int score)
        {
            this.score = score;
        }

    }

    public void syncVideo()
    {
        delay();

        Map<String, Set<String>> video_tags_update = new HashMap<>();
        Map<String, List<VideoNode>> tag_videos_update = new HashMap<>();

        EntityManager hot_em = null;

        try
        {
            hot_em = hotEntityManagerFactory.createEntityManager();

            long begin_time = Instant.now().toEpochMilli();

            List<VideoNode> nodes = ((List<?>) hot_em
                    .createNativeQuery(
                            "select xid, mtag, score from hot_item_recommend"
                                    + " where type=:type")
                    .setParameter("type", MtInfoType.VIDEO).getResultList())
                            .stream().map(r -> {
                                Object[] row = (Object[]) r;
                                int col = 0;
                                String xid = Optional
                                        .ofNullable((String) row[col++])
                                        .map(String::trim).orElse("");
                                String mtag = Optional
                                        .ofNullable((String) row[col++])
                                        .map(String::trim).orElse("");
                                int score = Optional
                                        .ofNullable((Integer) row[col++])
                                        .orElse(0);
                                if (xid.isEmpty() || mtag.isEmpty())
                                    return null;
                                VideoNode node = new VideoNode();
                                node.setXid(xid);
                                node.setMtag(mtag);
                                node.setScore(score);
                                return node;
                            }).filter(Objects::nonNull)
                            .collect(Collectors.toList());
            video_tags_update = nodes.stream()
                    .collect(Collectors.groupingBy(VideoNode::getXid, Collectors
                            .mapping(VideoNode::getMtag, Collectors.toSet())));
            tag_videos_update = nodes.stream()
                    .collect(Collectors.groupingBy(VideoNode::getMtag));

            logger.info("sync video cost {} ms, num({}), tags({})",
                    Instant.now().toEpochMilli() - begin_time, nodes.size(),
                    video_tags_update.size());
            logger.debug("tag_videos: {}", tag_videos_update.entrySet().stream()
                    .map(entry -> entry.getKey() + ": "
                            + entry.getValue().stream().map(VideoNode::getXid)
                                    .collect(Collectors.joining(",")))
                    .collect(Collectors.joining(" | ")));
        }
        catch (Exception e)
        {
            logger.error("sync fail: {}", e);
            return;
        }
        finally
        {
            if (hot_em != null)
                hot_em.close();
        }

        try
        {
            data_lock.writeLock().lock();

            video_tags = video_tags_update;
            tag_videos = tag_videos_update;
        }
        catch (Exception e)
        {
            logger.error("lock fail: {}", e);
            return;
        }
        finally
        {
            data_lock.writeLock().unlock();
        }
    }

    /* ========== Cid ========== */
    private final int CID_EXPIRE_MINUTES = 15;
    private final int XID_EXPIRE_MINUTES = 3;
    private final int READ_EXPIRE_HOURS = 7 * 24;
    private final int DISLIKE_EXPIRE_HOURS = 2;

    class CidReadLog
    {
        private String xid;
        private Instant time;
        private int last_page;

        public CidReadLog(String xid, Instant time, int last_page)
        {
            super();
            this.xid = xid;
            this.time = time;
            this.last_page = last_page;
        }

        public String getXid()
        {
            return xid;
        }

        public void setXid(String xid)
        {
            this.xid = xid;
        }

        public Instant getTime()
        {
            return time;
        }

        public void setTime(Instant time)
        {
            this.time = time;
        }

        public int getLast_page()
        {
            return last_page;
        }

        public void setLast_page(int last_page)
        {
            this.last_page = last_page;
        }

    }

    class CidNode
    {
        private Instant access_time = Instant.now();
        private Instant flush_time = Instant.ofEpochSecond(0);

        private Instant hotlist_sync_time = Instant.ofEpochSecond(0);
        // fid -> { xid -> score }
        private Map<Integer, Map<String, Integer>> hotlist = new HashMap<>();

        private Instant videolist_sync_time = Instant.ofEpochSecond(0);
        // xid -> score
        private Map<String, Integer> videolist = new HashMap<>();

        private Instant topic_hots_sync_time = Instant.ofEpochSecond(0);
        // fid -> { xid -> score }
        private Map<Integer, Map<String, Integer>> topic_hots = new HashMap<>();

        // xid -> 创建时间
        private Map<String, Instant> read_xids = new HashMap<>();
        // xid -> 创建时间
        private Map<String, Instant> dislike_xids = new HashMap<>();

        private List<CidReadLog> read_log = new ArrayList<>();

        public Instant getTopic_hots_sync_time()
        {
            return topic_hots_sync_time;
        }

        public void setTopic_hots_sync_time(Instant topic_hots_sync_time)
        {
            this.topic_hots_sync_time = topic_hots_sync_time;
        }

        public Map<Integer, Map<String, Integer>> getTopic_hots()
        {
            return topic_hots;
        }

        public void setTopic_hots(Map<Integer, Map<String, Integer>> topic_hots)
        {
            this.topic_hots = topic_hots;
        }

        public Instant getHotlist_sync_time()
        {
            return hotlist_sync_time;
        }

        public void setHotlist_sync_time(Instant hotlist_sync_time)
        {
            this.hotlist_sync_time = hotlist_sync_time;
        }

        public Map<Integer, Map<String, Integer>> getHotlist()
        {
            return hotlist;
        }

        public void setHotlist(Map<Integer, Map<String, Integer>> hotlist)
        {
            this.hotlist = hotlist;
        }

        public Instant getVideolist_sync_time()
        {
            return videolist_sync_time;
        }

        public void setVideolist_sync_time(Instant videolist_sync_time)
        {
            this.videolist_sync_time = videolist_sync_time;
        }

        public Map<String, Integer> getVideolist()
        {
            return videolist;
        }

        public void setVideolist(Map<String, Integer> videolist)
        {
            this.videolist = videolist;
        }

        public Instant getAccess_time()
        {
            return access_time;
        }

        public void setAccess_time(Instant access_time)
        {
            this.access_time = access_time;
        }

        public Instant getFlush_time()
        {
            return flush_time;
        }

        public void setFlush_time(Instant flush_time)
        {
            this.flush_time = flush_time;
        }

        public Map<String, Instant> getRead_xids()
        {
            return read_xids;
        }

        public void setRead_xids(Map<String, Instant> read_xids)
        {
            this.read_xids = read_xids;
        }

        public Map<String, Instant> getDislike_xids()
        {
            return dislike_xids;
        }

        public void setDislike_xids(Map<String, Instant> dislike_xids)
        {
            this.dislike_xids = dislike_xids;
        }

        public List<CidReadLog> getRead_log()
        {
            return read_log;
        }

        public void setRead_log(List<CidReadLog> read_log)
        {
            this.read_log = read_log;
        }

    }

    class CidHash
    {
        private ReadWriteLock lock = new ReentrantReadWriteLock();

        private Map<Integer, CidNode> cid_nodes = new HashMap<>();

        public ReadWriteLock getLock()
        {
            return lock;
        }

        public void setLock(ReadWriteLock lock)
        {
            this.lock = lock;
        }

        public Map<Integer, CidNode> getCid_nodes()
        {
            return cid_nodes;
        }

        public void setCid_nodes(Map<Integer, CidNode> cid_nodes)
        {
            this.cid_nodes = cid_nodes;
        }

    }

    private final int CID_HASH_NUM = 1000;
    private CidHash[] cid_hashes;
    /* ========== Cid ========== */

    private int get_cid_hash_index(int cid)
    {
        return (int) ((String.valueOf(cid).hashCode() & 0xFFFFFFFFL)
                % CID_HASH_NUM);
    }

    private String get_cid_read_storage(int cid)
    {
        return String.format("%s/%d/%d-read", appconfig.storage.path,
                get_cid_hash_index(cid), cid);
    }

    // private String get_cid_read_log_storage(int cid)
    // {
    // return String.format("%s/%d/%d-read-log", appconfig.storage.path,
    // get_cid_hash_index(cid), cid);
    // }

    private String get_cid_dislike_storage(int cid)
    {
        return String.format("%s/%d/%d-dislike", appconfig.storage.path,
                get_cid_hash_index(cid), cid);
    }

    private int purge_hash_index = 0;

    class Flush
    {
        private int cid;
        private Map<String, Long> read_xids;
        private Map<String, Long> dislike_xids;
        private List<String> read_log;

        public int getCid()
        {
            return cid;
        }

        public void setCid(int cid)
        {
            this.cid = cid;
        }

        public Map<String, Long> getRead_xids()
        {
            return read_xids;
        }

        public void setRead_xids(Map<String, Long> read_xids)
        {
            this.read_xids = read_xids;
        }

        public Map<String, Long> getDislike_xids()
        {
            return dislike_xids;
        }

        public void setDislike_xids(Map<String, Long> dislike_xids)
        {
            this.dislike_xids = dislike_xids;
        }

        public List<String> getRead_log()
        {
            return read_log;
        }

        public void setRead_log(List<String> read_log)
        {
            this.read_log = read_log;
        }

    }

    public void purgeCid()
    {
        List<Flush> flushes = new ArrayList<>();
        Instant now = Instant.now();
        int purge_count = 0;
        int hotlist_delete_count = 0;
        int videolist_delete_count = 0;
        DateTimeFormatter formatter = DateTimeFormatter
                .ofPattern("yyyy-MM-dd HH:mm:ss");

        CidHash cid_hash = cid_hashes[purge_hash_index];
        try
        {
            cid_hash.getLock().writeLock().lock();

            for (Iterator<Entry<Integer, CidNode>> cids_it = cid_hash
                    .getCid_nodes().entrySet().iterator(); cids_it.hasNext();)
            {
                Entry<Integer, CidNode> cid_entry = cids_it.next();
                int cid = cid_entry.getKey();
                CidNode cid_node = cid_entry.getValue();

                Map<String, Instant> read_xids = cid_node.getRead_xids();
                Map<String, Instant> dislike_xids = cid_node.getDislike_xids();

                // 清理已读、不喜欢列表
                purge_read_dislike(now, read_xids, dislike_xids);

                // 回写存储
                if (cid_node.getAccess_time().isAfter(cid_node.getFlush_time()))
                {
                    Flush flush = new Flush();
                    flush.setCid(cid);
                    flush.setRead_xids(read_xids.entrySet().stream()
                            .collect(Collectors.toMap(Entry::getKey,
                                    entry -> entry.getValue().toEpochMilli())));
                    flush.setDislike_xids(dislike_xids.entrySet().stream()
                            .collect(Collectors.toMap(Entry::getKey,
                                    entry -> entry.getValue().toEpochMilli())));
                    flush.setRead_log(cid_node.getRead_log().stream()
                            .map(log -> String.format("%s|%s|%d\n",
                                    log.getXid(),
                                    LocalDateTime
                                            .ofInstant(log.getTime(),
                                                    ZoneId.systemDefault())
                                            .format(formatter),
                                    log.getLast_page()))
                            .collect(Collectors.toList()));
                    flushes.add(flush);

                    cid_node.setRead_log(new ArrayList<>());
                    cid_node.setFlush_time(now);
                }

                if (cid_node.getAccess_time().until(now,
                        ChronoUnit.MINUTES) >= CID_EXPIRE_MINUTES)
                { // 清理不活跃用户
                    cids_it.remove();
                    purge_count++;
                }
                // else
                // {
                // 推荐列表中删除已读
                // for (Map<String, Integer> xids : cid_node.getHotlist()
                // .values())
                // {
                // for (Iterator<Entry<String, Integer>> xids_it = xids
                // .entrySet().iterator(); xids_it.hasNext();)
                // {
                // Entry<String, Integer> xid_entry = xids_it.next();
                // String xid = xid_entry.getKey();
                // if (read_xids.containsKey(xid))
                // {
                // xids_it.remove();
                // hotlist_delete_count++;
                // }
                // }
                // }

                // 视频列表中删除已读
                // for (Iterator<Entry<String, Integer>> xids_it = cid_node
                // .getVideolist().entrySet().iterator(); xids_it
                // .hasNext();)
                // {
                // Entry<String, Integer> xid_entry = xids_it.next();
                // String xid = xid_entry.getKey();
                // if (read_xids.containsKey(xid))
                // {
                // xids_it.remove();
                // videolist_delete_count++;
                // }
                // }
                // }
            }
        }
        catch (Exception e)
        {
            logger.error("fail: ", e);
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }

        if (!flushes.isEmpty())
        {
            long flush_begin_time = Instant.now().toEpochMilli();
            for (Flush flush : flushes)
            {
                String read_xids_s;
                try
                {
                    read_xids_s = objectMapper
                            .writeValueAsString(flush.getRead_xids());
                }
                catch (JsonProcessingException e1)
                {
                    read_xids_s = "{}";
                }
                String dislike_xids_s;
                try
                {
                    dislike_xids_s = objectMapper
                            .writeValueAsString(flush.getDislike_xids());
                }
                catch (JsonProcessingException e1)
                {
                    dislike_xids_s = "{}";
                }
                try (BufferedWriter read_writer = new BufferedWriter(
                        new FileWriter(get_cid_read_storage(flush.getCid())));
                        BufferedWriter dislike_writer = new BufferedWriter(
                                new FileWriter(get_cid_dislike_storage(
                                        flush.getCid())));
                // BufferedWriter read_log_writer = new BufferedWriter(
                // new FileWriter(get_cid_read_log_storage(
                // flush.getCid()), true))
                )
                {
                    read_writer.write(read_xids_s);
                    dislike_writer.write(dislike_xids_s);
                    // for (String log : flush.getRead_log())
                    // {
                    // read_log_writer.write(log);
                    // }
                }
                catch (Exception e)
                {
                    logger.error("write fail: {}", e.getMessage());
                }
            }
            int cost = (int) (Instant.now().toEpochMilli() - flush_begin_time);
            logger.info(
                    "purge_hash_index({}): cost {} ms, purge({}), flush({}),"
                            + " hotlist delete({}), videolist delete({})",
                    purge_hash_index, cost, purge_count, flushes.size(),
                    hotlist_delete_count, videolist_delete_count);
        }

        if (++purge_hash_index == cid_hashes.length)
        {
            purge_hash_index = 0;
        }
    }

    private final int PAGE_SIZE = 5;

    private String post(String url, String request) throws Exception
    {
        HttpURLConnection conn = null;
        try
        {
            conn = (HttpURLConnection) new URL(url).openConnection();
            conn.setRequestMethod("POST");
            conn.setDoOutput(true);
            conn.setRequestProperty("User-Agent",
                    "Mozilla/5.0 (Windows NT 5.1)"
                            + " AppleWebKit/537.36 (KHTML, like Gecko)"
                            + " Chrome/30.0.1599.101 Safari/537.36");
            conn.setRequestProperty("Content-Type",
                    "application/json;charset=utf-8");
            conn.setRequestProperty("Accept-Language",
                    Locale.getDefault().toString());
            conn.setRequestProperty("Accept-Charset", "UTF-8");
            conn.setConnectTimeout(1 * 1000);
            conn.setReadTimeout(1 * 1000);
            conn.connect();

            try (OutputStream out = conn.getOutputStream())
            {
                logger.debug("request body({})", request);
                byte[] b = request.getBytes();
                out.write(b, 0, b.length);
                out.flush();
            }

            try (InputStream in = conn.getInputStream())
            {
                byte[] buf = local.get().getBuf();
                int size = 0;
                while (true)
                {
                    int n = in.read(buf, size, buf.length - size);
                    if (n == -1)
                        break;
                    size += n;
                }
                String response = new String(buf, 0, size, "UTF-8");
                logger.debug("response({})", response);
                return response;
            }
        }
        finally
        {
            if (conn != null)
                conn.disconnect();
        }
    }

    @Override
    public String hotList(int cid, int is_first, List<Integer> fids,
            String device, String agent)
    {
        try
        {
            long begin_time = Instant.now().toEpochMilli();
            HotListResponse response = _hotList(cid, is_first, fids, device,
                    agent);
            int cost = (int) (Instant.now().toEpochMilli() - begin_time);
            if (cost > 100)
            {
                logger.info("hotlist cost {} ms", cost);
            }
            response.set_cost(cost);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private HotListResponse _hotList(int cid, int is_first, List<Integer> fids,
            String device, String agent)
    {
        logger.debug(
                "cid({}): fids({}) cid_hash_index({})", cid, fids.stream()
                        .map(String::valueOf).collect(Collectors.joining(",")),
                get_cid_hash_index(cid));
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        boolean need_load = false;
        // List<Integer> fids_not_sync = fids;
        List<String> reads = new ArrayList<>();
        Instant now = Instant.now();

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);

                    if (cid_node.getHotlist_sync_time().until(now,
                            ChronoUnit.MINUTES) > XID_EXPIRE_MINUTES)
                    { // 列表过期
                        logger.debug("cid({}): expire", cid);
                    }
                    else
                    {
                        Set<Integer> fids_sync = cid_node.getHotlist().keySet();
                        List<Integer> fids_not_sync = fids.stream()
                                .filter(fid -> !fids_sync.contains(fid))
                                .collect(Collectors.toList());
                        if (fids_not_sync.isEmpty())
                        { // fids都已同步
                            FetchPageResult fetch_page_result = fetch_hot_page(
                                    cid_node, fids);
                            if (fetch_page_result.getPage().size() == PAGE_SIZE)
                            { // 当前数据即可满足
                              // 修改已读列表
                              // cid_node.setAccess_time(now);
                                modify_read(fetch_page_result.getPage(), now, 0,
                                        cid_node.getRead_xids(),
                                        cid_node.getRead_log());
                                logger.debug("cid({}): hit full page", cid);
                                HotListResponse response = new HotListResponse(
                                        0, "成功");
                                response.setList(fetch_page_result.getPage());
                                return response;
                            }
                            logger.debug("cid({}): not enough page", cid);
                        }
                    }
                    reads = filter_hot_read_xids(cid_node.getRead_xids(), fids);
                }
            }
            else
            {
                need_load = true;
                logger.debug("cid({}): need_load", cid);
            }
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // 载入cid
        Map<String, Instant> read_xids_load = new HashMap<>();
        Map<String, Instant> dislike_xids_load = new HashMap<>();
        if (need_load)
        {
            LoadCidResult load_cid_result = load_cid(cid);
            if (load_cid_result != null)
            {
                read_xids_load = load_cid_result.getRead_xids();
                dislike_xids_load = load_cid_result.getDislike_xids();
                reads = filter_hot_read_xids(read_xids_load, fids);
                logger.debug("cid({}): load, read({}), dislike({})", cid,
                        read_xids_load.size(), dislike_xids_load.size());
            }
        }

        // 同步xids
        SyncHotListResult sync_xids_result = sync_hotlist(cid, fids, device,
                agent, reads);

        try
        {
            cid_hash.getLock().writeLock().lock();

            long modify_begin_time = Instant.now().toEpochMilli();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node == null)
            {
                cid_node = new CidNode();
                cid_node.setRead_xids(read_xids_load);
                cid_node.setDislike_xids(dislike_xids_load);
                cid_hash.getCid_nodes().put(cid, cid_node);
                logger.debug("cid({}): add to hash", cid);
            }
            cid_node.setHotlist_sync_time(now);
            Map<Integer, Map<String, Integer>> hotlist = cid_node.getHotlist();
            hotlist.putAll(sync_xids_result.getHotlist());

            HotListResponse response = new HotListResponse(0, "成功");
            FetchPageResult fetch_page_result = fetch_hot_page(cid_node, fids);
            List<HotResponseNode> page = fetch_page_result.getPage();
            if (!page.isEmpty())
            {
                // 修改已读列表
                modify_read(fetch_page_result.getPage(), now, 0,
                        cid_node.getRead_xids(), cid_node.getRead_log());
                // cid_node.setAccess_time(now);
                response.setList(page);
                logger.debug("cid({}): sync page", cid);
            }
            else
            {
                if (is_first == 1)
                { // 首次进入无记录则显示最后一页
                    modify_read(fetch_page_result.getPage(), now, 1,
                            cid_node.getRead_xids(), cid_node.getRead_log());
                    response.setList(
                            fetch_last_hot_page(fids, cid_node.getRead_xids()));
                }
                else
                {
                    response.setList(new ArrayList<>());
                }
                logger.debug("cid({}): sync empty page, is_first({})", cid,
                        is_first);
            }

            int cost = (int) (Instant.now().toEpochMilli() - modify_begin_time);
            if (cost > 50)
            {
                logger.info("cid({}): modify cid_node cost {} ms", cid, cost);
            }

            return response;
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }
    }

    private void modify_read(List<HotResponseNode> page, Instant now,
            int last_page, Map<String, Instant> read_xids,
            List<CidReadLog> read_log)
    {
        int add = 0;
        for (HotResponseNode node : page)
        {
            Instant now_adjust = now.plusMillis(++add);
            read_xids.put(node.getXid(), now_adjust);
            read_log.add(new CidReadLog(node.getXid(), now_adjust, last_page));
        }
    }

    private List<String> filter_hot_read_xids(Map<String, Instant> read_xids,
            List<Integer> fids)
    {
        if (fids == null || fids.isEmpty() || fids.contains(0))
        {
            return read_xids.keySet().stream().collect(Collectors.toList());
        }

        try
        {
            data_lock.readLock().lock();

            return read_xids.keySet().stream().filter(xid -> {
                ContentNode cn = contents.get(xid);
                if (cn == null)
                    return false;
                if (fids.contains(cn.getFid()))
                    return false;
                return true;
            }).collect(Collectors.toList());
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private boolean purge_read_dislike(Instant now,
            Map<String, Instant> read_xids, Map<String, Instant> dislike_xids)
    {
        boolean is_purged = false;
        for (Iterator<Entry<String, Instant>> it = read_xids.entrySet()
                .iterator(); it.hasNext();)
        { // 删除过期已读列表
            Entry<String, Instant> entry = it.next();
            if (entry.getValue().until(now,
                    ChronoUnit.HOURS) >= READ_EXPIRE_HOURS)
            {
                it.remove();
                is_purged = true;
            }
        }
        for (Iterator<Entry<String, Instant>> it = dislike_xids.entrySet()
                .iterator(); it.hasNext();)
        { // 删除过期不喜欢列表
            Entry<String, Instant> entry = it.next();
            if (entry.getValue().until(now,
                    ChronoUnit.HOURS) >= DISLIKE_EXPIRE_HOURS)
            {
                it.remove();
                is_purged = true;
            }
        }
        return is_purged;
    }

    class FetchPageResult
    {
        private List<HotResponseNode> page = new ArrayList<>();

        public List<HotResponseNode> getPage()
        {
            return page;
        }

        public void setPage(List<HotResponseNode> page)
        {
            this.page = page;
        }

    }

    private FetchPageResult fetch_hot_page(CidNode cid_node, List<Integer> fids)
    {
        Map<Integer, Map<String, Integer>> hotlist = cid_node.getHotlist();
        List<String> hotlist_fids = fids.stream()
                .flatMap(fid -> Optional.ofNullable(hotlist.get(fid))
                        //// .orElse(Optional.ofNullable(hotlist_default.get(fid))
                        // .orElse(new HashMap<>()))
                        .orElse(new HashMap<>()).entrySet().stream())
                .sorted((x, y) -> -x.getValue().compareTo(y.getValue()))
                .map(Entry::getKey).collect(Collectors.toList());

        try
        {
            data_lock.readLock().lock();

            FetchPageResult result = new FetchPageResult();

            Map<String, Instant> read_xids = cid_node.getRead_xids();
            Map<String, Instant> dislike_xids = cid_node.getDislike_xids();
            // AtomicInteger no_content_count = new AtomicInteger(0);
            List<HotResponseNode> page = hotlist_fids.stream().map(xid -> {
                if (read_xids != null && read_xids.containsKey(xid))
                    return null;
                if (dislike_xids != null && dislike_xids.containsKey(xid))
                    return null;
                if (blacks.containsKey(xid))
                    return null;
                ContentNode cn = contents.get(xid);
                if (cn == null)
                {
                    // no_content_count.incrementAndGet();
                    return null;
                }
                HotResponseNode node = new HotResponseNode();
                node.setXid(xid);
                node.setContent(
                        Optional.ofNullable(cn.getContent()).orElse("{}"));
                return node;
            }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            // if (no_content_count.get() > 0)
            // {
            // logger.error("no_content({})", no_content_count.get());
            // }
            result.setPage(page);
            return result;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private List<HotResponseNode> fetch_last_hot_page(List<Integer> fids,
            Map<String, Instant> read_xids)
    {
        try
        {
            data_lock.readLock().lock();

            List<HotResponseNode> page = read_xids.entrySet().stream()
                    .sorted((x, y) -> -x.getValue().compareTo(y.getValue()))
                    .map(entry -> {
                        String xid = entry.getKey();
                        ContentNode cn = contents.get(xid);
                        if (cn == null)
                            return null;
                        if (!fids.isEmpty() && !fids.contains(0)
                                && !fids.contains(cn.getFid()))
                            return null;
                        HotResponseNode node = new HotResponseNode();
                        node.setXid(xid);
                        node.setContent(Optional.ofNullable(cn.getContent())
                                .orElse("{}"));
                        return node;
                    }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            Collections.reverse(page);
            return page;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    class LoadCidResult
    {
        private Map<String, Instant> read_xids = new HashMap<>();
        private Map<String, Instant> dislike_xids = new HashMap<>();

        public Map<String, Instant> getRead_xids()
        {
            return read_xids;
        }

        public void setRead_xids(Map<String, Instant> read_xids)
        {
            this.read_xids = read_xids;
        }

        public Map<String, Instant> getDislike_xids()
        {
            return dislike_xids;
        }

        public void setDislike_xids(Map<String, Instant> dislike_xids)
        {
            this.dislike_xids = dislike_xids;
        }

    }

    private LoadCidResult load_cid(int cid)
    {
        File read_file = new File(get_cid_read_storage(cid));
        File dislike_file = new File(get_cid_dislike_storage(cid));
        if (!read_file.exists() || !dislike_file.exists())
        {
            return null;
        }
        if (!read_file.isFile() || !read_file.canRead() || !read_file.canWrite()
                || !dislike_file.isFile() || !dislike_file.canRead()
                || !dislike_file.canWrite())
        {
            logger.error("storage({}) invalid", read_file.getPath());
            return null;
        }

        long read_begin_time = Instant.now().toEpochMilli();
        String read_xids_s;
        String dislike_xids_s;
        try (FileInputStream read_fis = new FileInputStream(read_file);
                FileInputStream dislike_fis = new FileInputStream(dislike_file))
        {
            byte[] local_buf = local.get().getBuf();
            logger.debug("local.buffer({})", local_buf.length);
            byte[] buf;
            int len;

            if (read_fis.available() > local_buf.length)
                buf = new byte[(int) read_fis.available()];
            else
                buf = local_buf;
            len = read_fis.read(buf);
            read_xids_s = new String(buf, 0, len);

            if (dislike_fis.available() > local_buf.length)
                buf = new byte[(int) dislike_fis.available()];
            else
                buf = local_buf;
            len = dislike_fis.read(buf);
            dislike_xids_s = new String(buf, 0, len);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            return null;
        }
        int cost = (int) (Instant.now().toEpochMilli() - read_begin_time);
        if (cost > 50)
        {
            logger.info("cid({}): load cid read cost {} ms", cid, cost);
        }

        LoadCidResult result = new LoadCidResult();

        long convert_begin_time = Instant.now().toEpochMilli();

        Map<String, Long> read_xids;
        try
        {
            read_xids = objectMapper.readValue(
                    Optional.ofNullable(read_xids_s).map(String::trim)
                            .filter(s -> !s.isEmpty()).orElse("{}"),
                    new TypeReference<HashMap<String, Long>>()
                    {
                    });
        }
        catch (Exception e)
        {
            logger.error("storage({}): read_xids invalid: {}",
                    read_file.getPath(), e.getMessage());
            read_xids = new HashMap<>();
        }
        result.setRead_xids(read_xids.entrySet().stream()
                .collect(Collectors.toMap(Entry::getKey,
                        entry -> Instant.ofEpochMilli(entry.getValue()))));

        Map<String, Long> dislike_xids;
        try
        {
            dislike_xids = objectMapper.readValue(
                    Optional.ofNullable(dislike_xids_s).map(String::trim)
                            .filter(s -> !s.isEmpty()).orElse("{}"),
                    new TypeReference<HashMap<String, Long>>()
                    {
                    });
        }
        catch (

        Exception e)
        {
            logger.error("storage({}): read_xids invalid: {}",
                    read_file.getPath(), e.getMessage());
            dislike_xids = new HashMap<>();
        }
        result.setDislike_xids(dislike_xids.entrySet().stream()
                .collect(Collectors.toMap(Entry::getKey,
                        entry -> Instant.ofEpochMilli(entry.getValue()))));

        purge_read_dislike(Instant.now(), result.getRead_xids(),
                result.getDislike_xids());

        cost = (int) (Instant.now().toEpochMilli() - convert_begin_time);
        if (cost > 50)
        {
            logger.info("cid({}): load cid convert cost {} ms", cid, cost);
        }

        return result;
    }

    class SyncHotListResult
    {
        private Map<Integer, Map<String, Integer>> hotlist = new HashMap<>();
        private long size;

        public long getSize()
        {
            return size;
        }

        public void setSize(long size)
        {
            this.size = size;
        }

        public Map<Integer, Map<String, Integer>> getHotlist()
        {
            return hotlist;
        }

        public void setHotlist(Map<Integer, Map<String, Integer>> hotlist)
        {
            this.hotlist = hotlist;
        }

    }

    private SyncHotListResult sync_hotlist(int cid, List<Integer> fids,
            String device, String agent, List<String> reads)
    {
        try
        {
            long http_begin_time = Instant.now().toEpochMilli();

            SyncHotListRequest request = new SyncHotListRequest();
            request.setCid(cid);
            if (fids != null && !fids.isEmpty() && !fids.contains(0))
            {
                request.setFids(fids);
            }
            if (reads != null && !reads.isEmpty())
            {
                request.setRead_xids(reads);
            }
            request.setDevice(Optional.ofNullable(device).orElse(""));
            request.setAgent(Optional.ofNullable(agent).orElse(""));
            String response_s = post(appconfig.http.hotlist,
                    objectMapper.writeValueAsString(request));
            SyncHotListResponse response = objectMapper.readValue(response_s,
                    SyncHotListResponse.class);
            SyncHotListResult result = new SyncHotListResult();
            result.setSize(response_s.getBytes().length);
            result.setHotlist(response.getData());

            int cost = (int) (Instant.now().toEpochMilli() - http_begin_time);
            // logger.debug(
            // "cid({}) fids({}): sync hotlist http cost {} ms,"
            // + " reads({}), num({}) size({})",
            // cid,
            // Optional.ofNullable(fids).orElse(new ArrayList<>()).stream()
            // .map(String::valueOf)
            // .collect(Collectors.joining(",")),
            // cost,
            // Optional.ofNullable(reads).orElse(new ArrayList<>()).size(),
            // result.getHotlist().size(), result.getSize());
            if (cost > 100)
            {
                logger.info(
                        "cid({}) fids({}): sync hotlist http cost {} ms,"
                                + " reads({}), num({}) size({})",
                        cid,
                        Optional.ofNullable(fids).orElse(new ArrayList<>())
                                .stream().map(String::valueOf)
                                .collect(Collectors.joining(",")),
                        cost,
                        Optional.ofNullable(reads).orElse(new ArrayList<>())
                                .size(),
                        result.getHotlist().values().stream()
                                .collect(Collectors.summingInt(Map::size)),
                        result.getSize());
            }

            return result;
        }
        catch (Exception e)
        {
            logger.info("cid({}) sync fail: {}", cid, e);
            return new SyncHotListResult();
        }
    }

    class DislikeStatus
    {
        public final static int UNPROCESSED = 0;
    }

    @Override
    public String dislike(int cid, String xid)
    {
        try
        {
            DislikeResponse response = _dislike(cid, xid);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private DislikeResponse _dislike(int cid, String xid)
    {
        Instant now = Instant.now();

        EntityManager hot_em = null;

        try
        {
            hot_em = hotEntityManagerFactory.createEntityManager();

            try
            {
                hot_em.createNativeQuery("select ctime from hot_client_dislike"
                        + " where cid=:cid and xid=:xid")
                        .setParameter("cid", cid).setParameter("xid", xid)
                        .getSingleResult();
            }
            catch (NoResultException e)
            {
                try
                {
                    hot_em.getTransaction().begin();

                    hot_em.createNativeQuery("insert into hot_client_dislike"
                            + " (cid, xid, status, ctime)"
                            + " values(:cid, :xid, :status, :ctime)")
                            .setParameter("cid", cid).setParameter("xid", xid)
                            .setParameter("status", DislikeStatus.UNPROCESSED)
                            .setParameter("ctime",
                                    Instant.now().getEpochSecond())
                            .executeUpdate();

                    hot_em.getTransaction().commit();
                }
                catch (Exception insert_e)
                {
                    logger.error("insert fail: {}", e);
                    try
                    {
                        if (hot_em.getTransaction().isActive())
                            hot_em.getTransaction().rollback();
                    }
                    catch (Exception rollback_e)
                    {
                        logger.error("rollback fail: {}", rollback_e);
                    }
                    throw insert_e;
                }
            }
        }
        finally
        {
            if (hot_em != null)
                hot_em.close();
        }

        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);
                    cid_node.getDislike_xids().put(xid, now);
                    return new DislikeResponse(0, "成功");
                }
            }
            else
            {
                String errmsg = String.format(
                        "cid(%s) not in buffer, abnormal" + ", should load",
                        cid);
                logger.error(errmsg);
                return new DislikeResponse(9999, errmsg);
            }
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }
    }

    @Override
    public String clearRead(int cid)
    {
        try
        {
            ClearReadResponse response = _clearRead(cid);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private ClearReadResponse _clearRead(int cid)
    {
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    cid_node.setRead_xids(new HashMap<>());
                }
            }
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e);
            return new ClearReadResponse(9999, e.getMessage());
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // try (BufferedWriter read_writer = new BufferedWriter(
        // new FileWriter(get_cid_read_storage(cid))))
        // {
        // read_writer.write("{}");
        // }
        // catch (Exception e)
        // {
        // logger.error("write fail: {}", e.getMessage());
        // }

        return new ClearReadResponse(0, "成功");
    }

    @Override
    public String getRead(int cid)
    {
        try
        {
            GetReadResponse response = _getRead(cid);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private GetReadResponse _getRead(int cid)
    {
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];
        DateTimeFormatter formatter = DateTimeFormatter
                .ofPattern("yyyy-MM-dd HH:mm:ss");

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    GetReadResponse response = new GetReadResponse(0, "成功");
                    response.setList(cid_node
                            .getRead_xids().entrySet().stream().sorted((x,
                                    y) -> x.getValue().compareTo(y.getValue()))
                            .map(entry -> {
                                GetReadResponseNode node = new GetReadResponseNode();
                                node.setXid(entry.getKey());
                                node.setRead_time(LocalDateTime
                                        .ofInstant(entry.getValue(),
                                                ZoneId.systemDefault())
                                        .format(formatter));
                                return node;
                            }).collect(Collectors.toList()));
                    logger.debug("cid({}): in buffer read_xids({})", cid,
                            cid_node.getRead_xids().size());
                    return response;
                }
            }
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e);
            return new GetReadResponse(9999, e.getMessage());
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        try (FileInputStream fis = new FileInputStream(
                get_cid_read_storage(cid)))
        {
            byte[] buf = new byte[(int) fis.available()];
            fis.read(buf);
            Map<String, Long> xids = objectMapper.readValue(
                    Optional.ofNullable(new String(buf)).map(String::trim)
                            .filter(s -> !s.isEmpty()).orElse("{}"),
                    new TypeReference<HashMap<String, Long>>()
                    {
                    });

            GetReadResponse response = new GetReadResponse(0, "成功");
            ZoneOffset zone_offset = ZonedDateTime.now().getOffset();
            response.setList(xids.entrySet().stream()
                    .sorted((x, y) -> x.getValue().compareTo(y.getValue()))
                    .map(entry -> {
                        GetReadResponseNode node = new GetReadResponseNode();
                        node.setXid(entry.getKey());
                        node.setRead_time(
                                LocalDateTime
                                        .ofEpochSecond(entry.getValue() / 1000,
                                                0, zone_offset)
                                        .format(formatter));
                        return node;
                    }).collect(Collectors.toList()));
            return response;
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            return new GetReadResponse(9999, e.getMessage());
        }
    }

    @Override
    public String getDislike(int cid)
    {
        try
        {
            GetDislikeResponse response = _getDislike(cid);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private GetDislikeResponse _getDislike(int cid)
    {
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    GetDislikeResponse response = new GetDislikeResponse(0,
                            "成功");
                    response.setList(cid_node.getDislike_xids().entrySet()
                            .stream()
                            .sorted((x, y) -> x.getValue()
                                    .compareTo(y.getValue()))
                            .map(Entry::getKey).collect(Collectors.toList()));
                    logger.debug("cid({}): in buffer dislike_xids({})", cid,
                            cid_node.getDislike_xids().size());
                    return response;
                }
            }
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e);
            return new GetDislikeResponse(9999, e.getMessage());
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        try (FileInputStream fis = new FileInputStream(
                get_cid_dislike_storage(cid)))
        {
            byte[] buf = new byte[(int) fis.available()];
            fis.read(buf);
            Map<String, Long> xids = objectMapper.readValue(
                    Optional.ofNullable(new String(buf)).map(String::trim)
                            .filter(s -> !s.isEmpty()).orElse("{}"),
                    new TypeReference<HashMap<String, Long>>()
                    {
                    });

            GetDislikeResponse response = new GetDislikeResponse(0, "成功");
            response.setList(xids.entrySet().stream()
                    .sorted((x, y) -> x.getValue().compareTo(y.getValue()))
                    .map(Entry::getKey).collect(Collectors.toList()));
            return response;
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            return new GetDislikeResponse(9999, e.getMessage());
        }
    }

    @Override
    public String getList(int cid)
    {
        try
        {
            GetListResponse response = _getList(cid);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private GetListResponse _getList(int cid)
    {
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        try
        {
            cid_hash.getLock().readLock().lock();

            GetListResponse response = new GetListResponse(0, "成功");
            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    try
                    {
                        data_lock.readLock().lock();

                        response.setList(Optional
                                .ofNullable(cid_node.getHotlist().get(0))
                                .orElse(new HashMap<>()).keySet().stream()
                                .map(xid -> {
                                    GetListResponseNode node = new GetListResponseNode();
                                    node.setXid(xid);
                                    ContentNode cn = contents.get(xid);
                                    if (cn == null)
                                    {
                                        node.setStatus(
                                                GetListResponseNode.Status.NO_CONTENT);
                                        return node;
                                    }
                                    node.setFid(cn.getFid());
                                    if (blacks.containsKey(xid))
                                    {
                                        node.setStatus(
                                                GetListResponseNode.Status.BLACK);
                                    }
                                    else if (cid_node.getRead_xids()
                                            .containsKey(xid))
                                    {
                                        node.setStatus(
                                                GetListResponseNode.Status.READ);
                                    }
                                    if (cid_node.getDislike_xids()
                                            .containsKey(xid))
                                    {
                                        node.setStatus(
                                                GetListResponseNode.Status.DISLIKE);
                                    }
                                    return node;
                                }).filter(Objects::nonNull)
                                .collect(Collectors.toList()));
                    }
                    finally
                    {
                        data_lock.readLock().unlock();
                    }
                }
            }
            return response;
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e);
            return new GetListResponse(9999, e.getMessage());
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }
    }

    // @Override
    // public String getStat()
    // {
    // try
    // {
    // GetStatResponse response = _getStat();
    // return objectMapper.writeValueAsString(response);
    // }
    // catch (Exception e)
    // {
    // logger.error("fail: {}", e.getMessage());
    // try
    // {
    // return objectMapper.writeValueAsString(
    // new BaseResponse(9999, e.getMessage()));
    // }
    // catch (JsonProcessingException e1)
    // {
    // return error_response;
    // }
    // }
    // }
    //
    // private GetStatResponse _getStat()
    // {
    // for (int i = 0; i < cid_hashes.length; i++)
    // {
    // CidHash cid_hash = cid_hashes[i];
    //
    // try
    // {
    // cid_hash.getLock().readLock().lock();
    //
    // }
    // finally
    // {
    // cid_hash.getLock().readLock().unlock();
    // }
    // }
    // GetStatResponse response = new GetStatResponse(0, "成功");
    // return response;
    // }

    @Override
    public String videoList(int cid, int is_first)
    {
        try
        {
            long begin_time = Instant.now().toEpochMilli();
            VideoListResponse response = _videoList(cid, is_first);
            int cost = (int) (Instant.now().toEpochMilli() - begin_time);
            if (cost > 100)
            {
                logger.info("videolist cost {} ms", cost);
            }
            response.set_cost(cost);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private VideoListResponse _videoList(int cid, int is_first)
    {
        logger.debug("cid({}): cid_hash_index({})", cid,
                get_cid_hash_index(cid));
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        boolean need_load = false;
        List<String> reads = new ArrayList<>();
        Instant now = Instant.now();

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);

                    if (cid_node.getVideolist_sync_time().until(now,
                            ChronoUnit.MINUTES) > XID_EXPIRE_MINUTES)
                    { // 列表过期
                        logger.debug("cid({}): expire", cid);
                    }
                    else
                    {
                        FetchPageResult fetch_page_result = fetch_video_page(
                                cid_node);
                        if (fetch_page_result.getPage().size() == PAGE_SIZE)
                        { // 当前数据即可满足
                          // 修改已读列表
                            modify_read(fetch_page_result.getPage(), now, 0,
                                    cid_node.getRead_xids(),
                                    cid_node.getRead_log());
                            logger.debug("cid({}): hit full page", cid);
                            VideoListResponse response = new VideoListResponse(
                                    0, "成功");
                            response.setList(fetch_page_result.getPage());
                            logger.debug("cid({}): video [ {} ]", cid,
                                    fetch_page_result.getPage().stream()
                                            .map(HotResponseNode::getXid)
                                            .collect(Collectors.joining(", ")));
                            return response;
                        }
                        logger.debug("cid({}): not enough page", cid);
                    }
                    reads = filter_video_read_xids(cid_node.getRead_xids());
                }
            }
            else
            {
                need_load = true;
                logger.debug("cid({}): need_load", cid);
            }
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // 载入cid
        Map<String, Instant> read_xids_load = new HashMap<>();
        Map<String, Instant> dislike_xids_load = new HashMap<>();
        if (need_load)
        {
            LoadCidResult load_cid_result = load_cid(cid);
            if (load_cid_result != null)
            {
                read_xids_load = load_cid_result.getRead_xids();
                dislike_xids_load = load_cid_result.getDislike_xids();
                reads = filter_video_read_xids(read_xids_load);
                logger.debug("cid({}): load, read({}), dislike({})", cid,
                        read_xids_load.size(), dislike_xids_load.size());
            }
        }

        // 同步xids
        SyncVideoListResult sync_videolist_result = sync_videolist(cid, reads);

        try
        {
            cid_hash.getLock().writeLock().lock();

            long modify_begin_time = Instant.now().toEpochMilli();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node == null)
            {
                cid_node = new CidNode();
                cid_node.setRead_xids(read_xids_load);
                cid_node.setDislike_xids(dislike_xids_load);
                cid_hash.getCid_nodes().put(cid, cid_node);
                logger.debug("cid({}): add to hash", cid);
            }
            cid_node.setVideolist_sync_time(now);
            cid_node.setVideolist(sync_videolist_result.getVideolist());

            VideoListResponse response = new VideoListResponse(0, "成功");
            FetchPageResult fetch_page_result = fetch_video_page(cid_node);
            List<HotResponseNode> page = fetch_page_result.getPage();
            if (!page.isEmpty())
            {
                // 修改已读列表
                modify_read(fetch_page_result.getPage(), now, 0,
                        cid_node.getRead_xids(), cid_node.getRead_log());
                // cid_node.setAccess_time(now);
                response.setList(page);
                logger.debug("cid({}): sync page", cid);
            }
            else
            {
                if (is_first == 1)
                { // 首次进入无记录则显示最后一页
                    modify_read(fetch_page_result.getPage(), now, 1,
                            cid_node.getRead_xids(), cid_node.getRead_log());
                    response.setList(
                            fetch_last_video_page(cid_node.getRead_xids()));
                }
                else
                {
                    response.setList(new ArrayList<>());
                }
                logger.debug("cid({}): sync empty page, is_first({})", cid,
                        is_first);
            }
            logger.debug("cid({}): video [ {} ]", cid,
                    fetch_page_result.getPage().stream()
                            .map(HotResponseNode::getXid)
                            .collect(Collectors.joining(", ")));

            int cost = (int) (Instant.now().toEpochMilli() - modify_begin_time);
            if (cost > 50)
            {
                logger.info("cid({}): modify cid_node cost {} ms", cid, cost);
            }

            return response;
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }
    }

    private List<String> filter_video_read_xids(Map<String, Instant> read_xids)
    {
        try
        {
            data_lock.readLock().lock();

            return read_xids.keySet().stream().filter(xid -> {
                ContentNode cn = contents.get(xid);
                if (cn == null)
                    return false;
                if (!cn.getType().equals(MtInfoType.VIDEO))
                    return false;
                return true;
            }).collect(Collectors.toList());
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private FetchPageResult fetch_video_page(CidNode cid_node)
    {
        Map<String, Integer> videolist = cid_node.getVideolist();
        List<String> videolist_sort = videolist.entrySet().stream()
                .sorted(Collections
                        .reverseOrder(Comparator.comparingInt(Entry::getValue)))
                .map(Entry::getKey).collect(Collectors.toList());

        try
        {
            data_lock.readLock().lock();

            FetchPageResult result = new FetchPageResult();

            Map<String, Instant> read_xids = cid_node.getRead_xids();
            Map<String, Instant> dislike_xids = cid_node.getDislike_xids();
            // AtomicInteger no_content_count = new AtomicInteger(0);
            List<HotResponseNode> page = videolist_sort.stream().map(xid -> {
                if (read_xids != null && read_xids.containsKey(xid))
                    return null;
                if (dislike_xids != null && dislike_xids.containsKey(xid))
                    return null;
                if (blacks.containsKey(xid))
                    return null;
                ContentNode cn = contents.get(xid);
                if (cn == null)
                {
                    // no_content_count.incrementAndGet();
                    return null;
                }
                HotResponseNode node = new HotResponseNode();
                node.setXid(xid);
                node.setContent(
                        Optional.ofNullable(cn.getContent()).orElse("{}"));
                return node;
            }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            // if (no_content_count.get() > 0)
            // {
            // logger.error("no_content({})", no_content_count.get());
            // }

            result.setPage(page);
            return result;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private List<HotResponseNode> fetch_last_video_page(
            Map<String, Instant> read_xids)
    {
        try
        {
            data_lock.readLock().lock();

            List<HotResponseNode> page = read_xids.entrySet().stream()
                    .sorted((x, y) -> -x.getValue().compareTo(y.getValue()))
                    .map(entry -> {
                        String xid = entry.getKey();
                        ContentNode cn = contents.get(xid);
                        if (cn == null)
                            return null;
                        if (!cn.getType().equals(MtInfoType.VIDEO))
                            return null;
                        HotResponseNode node = new HotResponseNode();
                        node.setXid(xid);
                        node.setContent(Optional.ofNullable(cn.getContent())
                                .orElse("{}"));
                        return node;
                    }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            Collections.reverse(page);
            return page;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    class SyncVideoListResult
    {
        private Map<String, Integer> videolist = new HashMap<>();
        private long size;

        public long getSize()
        {
            return size;
        }

        public void setSize(long size)
        {
            this.size = size;
        }

        public Map<String, Integer> getVideolist()
        {
            return videolist;
        }

        public void setVideolist(Map<String, Integer> videolist)
        {
            this.videolist = videolist;
        }

    }

    private SyncVideoListResult sync_videolist(int cid, List<String> reads)
    {
        try
        {
            long http_begin_time = Instant.now().toEpochMilli();

            SyncVideoListRequest request = new SyncVideoListRequest();
            request.setCid(cid);
            if (reads != null && !reads.isEmpty())
            {
                request.setRead_xids(reads);
            }
            String response_s = post(appconfig.http.videolist,
                    objectMapper.writeValueAsString(request));
            SyncVideoListResponse response = objectMapper.readValue(response_s,
                    SyncVideoListResponse.class);
            SyncVideoListResult result = new SyncVideoListResult();
            result.setSize(response_s.getBytes().length);
            result.setVideolist(response.getData());

            int cost = (int) (Instant.now().toEpochMilli() - http_begin_time);
            // logger.debug(
            // "cid({}) reads({}): sync videolist http cost {} ms,"
            // + " num({}) size({})",
            // cid,
            // Optional.ofNullable(reads).orElse(new ArrayList<>()).size(),
            // cost, result.getVideolist().size(), result.getSize());
            if (cost > 100)
            {
                logger.info(
                        "cid({}) reads({}): sync videolist http cost {} ms,"
                                + " num({}) size({})",
                        cid,
                        Optional.ofNullable(reads).orElse(new ArrayList<>())
                                .size(),
                        cost, result.getVideolist().size(), result.getSize());
            }

            return result;
        }
        catch (Exception e)
        {
            logger.info("cid({}) sync fail: {}", cid, e.getMessage());
            return new SyncVideoListResult();
        }
    }

    private final int RELATED_VIDEO_PAGE_SIZE = 3;

    @Override
    public String relatedVideoList(int cid, String xid)
    {
        try
        {
            long begin_time = Instant.now().toEpochMilli();
            RelatedVideoListResponse response = _relatedVideoList(cid, xid);
            int cost = (int) (Instant.now().toEpochMilli() - begin_time);
            if (cost > 100)
            {
                logger.info("relatedVideoList cost {} ms", cost);
            }
            response.set_cost(cost);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private RelatedVideoListResponse _relatedVideoList(int cid, String xid)
            throws Exception
    {
        logger.debug("cid({}): cid_hash_index({})", cid,
                get_cid_hash_index(cid));
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        Instant now = Instant.now();
        boolean need_load = false;
        List<String> reads = new ArrayList<>();

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);

                    if (cid_node.getVideolist_sync_time().until(now,
                            ChronoUnit.MINUTES) > XID_EXPIRE_MINUTES)
                    { // 列表过期
                        logger.debug("cid({}): expire", cid);
                    }
                    else
                    {
                        FetchRelatedVideoPageResult fetch_page_result = fetch_related_video_page(
                                xid, cid_node);
                        // 修改已读列表
                        modify_read(fetch_page_result.getPage(), now, 0,
                                cid_node.getRead_xids(),
                                cid_node.getRead_log());
                        logger.debug("cid({}) xid({}): related video [ {} ]",
                                cid, xid,
                                fetch_page_result.getPage().stream()
                                        .map(HotResponseNode::getXid)
                                        .collect(Collectors.joining(", ")));
                        RelatedVideoListResponse response = new RelatedVideoListResponse(
                                0, "成功");
                        response.setCurr(fetch_page_result.getCurr());
                        response.setList(fetch_page_result.getPage());
                        return response;
                    }
                    reads = filter_video_read_xids(cid_node.getRead_xids());
                }
            }
            else
            {
                need_load = true;
                logger.debug("cid({}): need_load", cid);
            }
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // 载入cid
        Map<String, Instant> read_xids_load = new HashMap<>();
        Map<String, Instant> dislike_xids_load = new HashMap<>();
        if (need_load)
        {
            LoadCidResult load_cid_result = load_cid(cid);
            if (load_cid_result != null)
            {
                read_xids_load = load_cid_result.getRead_xids();
                dislike_xids_load = load_cid_result.getDislike_xids();
                reads = filter_video_read_xids(read_xids_load);
                logger.debug("cid({}): load, read({}), dislike({})", cid,
                        read_xids_load.size(), dislike_xids_load.size());
            }
        }

        // 同步xids
        SyncVideoListResult sync_videolist_result = sync_videolist(cid, reads);

        try
        {
            cid_hash.getLock().writeLock().lock();

            long modify_begin_time = Instant.now().toEpochMilli();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node == null)
            {
                cid_node = new CidNode();
                cid_node.setRead_xids(read_xids_load);
                cid_node.setDislike_xids(dislike_xids_load);
                cid_hash.getCid_nodes().put(cid, cid_node);
                logger.debug("cid({}): add to hash", cid);
            }
            cid_node.setVideolist_sync_time(now);
            cid_node.setVideolist(sync_videolist_result.getVideolist());

            FetchRelatedVideoPageResult fetch_page_result = fetch_related_video_page(
                    xid, cid_node);
            // 修改已读列表
            modify_read(fetch_page_result.getPage(), now, 0,
                    cid_node.getRead_xids(), cid_node.getRead_log());
            RelatedVideoListResponse response = new RelatedVideoListResponse(0,
                    "成功");
            response.setCurr(fetch_page_result.getCurr());
            response.setList(fetch_page_result.getPage());
            logger.debug("cid({}): related video [ {} ]", cid,
                    fetch_page_result.getPage().stream()
                            .map(HotResponseNode::getXid)
                            .collect(Collectors.joining(", ")));

            int cost = (int) (Instant.now().toEpochMilli() - modify_begin_time);
            if (cost > 50)
            {
                logger.info("cid({}): modify cid_node cost {} ms", cid, cost);
            }

            return response;
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }
    }

    class FetchRelatedVideoPageResult extends FetchPageResult
    {
        private HotResponseNode curr;

        public HotResponseNode getCurr()
        {
            return curr;
        }

        public void setCurr(HotResponseNode curr)
        {
            this.curr = curr;
        }

    }

    private FetchRelatedVideoPageResult fetch_related_video_page(String xid,
            CidNode cid_node) throws Exception
    {
        Map<String, Integer> videolist = cid_node.getVideolist();
        List<String> videolist_sort = videolist.entrySet().stream()
                .sorted(Collections
                        .reverseOrder(Comparator.comparingInt(Entry::getValue)))
                .map(Entry::getKey).collect(Collectors.toList());

        try
        {
            data_lock.readLock().lock();

            FetchRelatedVideoPageResult result = new FetchRelatedVideoPageResult();

            // 当前视频
            ContentNode content_node = contents.get(xid);
            if (content_node == null)
            {
                String errmsg = String.format("xid(%s) not in content, bad",
                        xid);
                logger.error(errmsg);
            }
            HotResponseNode curr = new HotResponseNode();
            curr.setXid(xid);
            curr.setContent(Optional.ofNullable(content_node).map(
                    cn -> Optional.ofNullable(cn.getContent()).orElse("{}"))
                    .orElse("{}"));
            result.setCurr(curr);

            // 相关视频
            Set<String> tags = Optional.ofNullable(video_tags.get(xid))
                    .orElse(new HashSet<>());
            if (tags.isEmpty())
            {
                logger.error("video_tags({}->{}), impossible", xid,
                        tags.size());
                return result;
            }
            Map<String, Integer> tag_xids = tags.stream()
                    .flatMap(tag -> Optional.ofNullable(tag_videos.get(tag))
                            .orElse(new ArrayList<>()).stream())
                    .map(VideoNode::getXid).distinct()
                    .collect(Collectors.toMap(x -> x, x -> 1));

            Map<String, Instant> read_xids = cid_node.getRead_xids();
            Map<String, Instant> dislike_xids = cid_node.getDislike_xids();
            // AtomicInteger no_content_count = new AtomicInteger(0);
            List<HotResponseNode> page = videolist_sort.stream().map(r_xid -> {
                if (!tag_xids.containsKey(r_xid))
                    return null;
                if (read_xids != null && read_xids.containsKey(r_xid))
                    return null;
                if (dislike_xids != null && dislike_xids.containsKey(r_xid))
                    return null;
                if (blacks.containsKey(r_xid))
                    return null;
                ContentNode cn = contents.get(r_xid);
                if (cn == null)
                {
                    // no_content_count.incrementAndGet();
                    return null;
                }
                HotResponseNode n = new HotResponseNode();
                n.setXid(r_xid);
                n.setContent(Optional.ofNullable(cn.getContent()).orElse("{}"));
                return n;
            }).filter(Objects::nonNull).limit(RELATED_VIDEO_PAGE_SIZE)
                    .collect(Collectors.toList());
            // if (no_content_count.get() > 0)
            // {
            // logger.error("xid({}): related no_content({})", xid,
            // no_content_count.get());
            // }
            result.setPage(page);
            return result;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    @Override
    public String unread(int cid, List<String> xids)
    {
        try
        {
            long begin_time = Instant.now().toEpochMilli();
            UnreadResponse response = _unread(cid, xids);
            int cost = (int) (Instant.now().toEpochMilli() - begin_time);
            if (cost > 100)
            {
                logger.info("relatedVideoList cost {} ms", cost);
            }
            response.set_cost(cost);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private UnreadResponse _unread(int cid, List<String> xids)
    {
        logger.debug("cid({}): cid_hash_index({})", cid,
                get_cid_hash_index(cid));
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        Instant now = Instant.now();

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);

                    Map<String, Instant> read_xids = cid_node.getRead_xids();
                    xids.forEach(xid -> {
                        read_xids.remove(xid);
                    });

                    return new UnreadResponse(0, "成功");
                }
            }
            logger.debug("cid({}): need_load", cid);
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // 载入cid
        Map<String, Instant> read_xids_load = new HashMap<>();
        Map<String, Instant> dislike_xids_load = new HashMap<>();
        LoadCidResult load_cid_result = load_cid(cid);
        if (load_cid_result != null)
        {
            read_xids_load = load_cid_result.getRead_xids();

            dislike_xids_load = load_cid_result.getDislike_xids();
            logger.debug("cid({}): load, read({}), dislike({})", cid,
                    read_xids_load.size(), dislike_xids_load.size());
        }

        try
        {
            cid_hash.getLock().writeLock().lock();

            long modify_begin_time = Instant.now().toEpochMilli();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node == null)
            {
                cid_node = new CidNode();
                cid_node.setRead_xids(read_xids_load);
                cid_node.setDislike_xids(dislike_xids_load);
                cid_hash.getCid_nodes().put(cid, cid_node);
                logger.debug("cid({}): add to hash", cid);
            }
            Map<String, Instant> read_xids = cid_node.getRead_xids();
            xids.forEach(xid -> {
                read_xids.remove(xid);
            });

            int cost = (int) (Instant.now().toEpochMilli() - modify_begin_time);
            if (cost > 50)
            {
                logger.info("cid({}): modify cid_node cost {} ms", cid, cost);
            }

            return new UnreadResponse(0, "成功");
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }
    }

    @Override
    public String hotListByTopic(int cid, int is_first, List<Integer> topics,
            String device, String agent)
    {
        try
        {
            long begin_time = Instant.now().toEpochMilli();
            HotListByTopicResponse response = _hotListByTopic(cid, is_first,
                    topics, device, agent);
            int cost = (int) (Instant.now().toEpochMilli() - begin_time);
            if (cost > 100)
            {
                logger.info("hotlistbytopic cost {} ms", cost);
            }
            response.set_cost(cost);
            return objectMapper.writeValueAsString(response);
        }
        catch (Exception e)
        {
            logger.error("fail: {}", e.getMessage());
            try
            {
                return objectMapper.writeValueAsString(
                        new BaseResponse(9999, e.getMessage()));
            }
            catch (JsonProcessingException e1)
            {
                return error_response;
            }
        }
    }

    private HotListByTopicResponse _hotListByTopic(int cid, int is_first,
            List<Integer> topics, String device, String agent)
    {
        logger.debug(
                "cid({}): topics({}) cid_hash_index({})", cid, topics.stream()
                        .map(String::valueOf).collect(Collectors.joining(",")),
                get_cid_hash_index(cid));
        CidHash cid_hash = cid_hashes[get_cid_hash_index(cid)];

        boolean need_load = false;
        List<String> reads = new ArrayList<>();
        Instant now = Instant.now();

        try
        {
            cid_hash.getLock().readLock().lock();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node != null)
            {
                synchronized (cid_node)
                {
                    // 用户活动
                    cid_node.setAccess_time(now);

                    if (cid_node.getTopic_hots_sync_time().until(now,
                            ChronoUnit.MINUTES) > XID_EXPIRE_MINUTES)
                    { // 列表过期
                        logger.debug("cid({}): expire", cid);
                    }
                    else
                    {
                        Set<Integer> topics_sync = cid_node.getTopic_hots()
                                .keySet();
                        List<Integer> topics_not_sync = topics.stream()
                                .filter(t -> !topics_sync.contains(t))
                                .collect(Collectors.toList());
                        if (topics_not_sync.isEmpty())
                        { // topics都已同步
                            FetchPageResult fetch_page_result = fetch_hot_page(
                                    cid_node, topics);
                            if (fetch_page_result.getPage().size() == PAGE_SIZE)
                            { // 当前数据即可满足
                              // 修改已读列表
                              // cid_node.setAccess_time(now);
                                modify_read(fetch_page_result.getPage(), now, 0,
                                        cid_node.getRead_xids(),
                                        cid_node.getRead_log());
                                logger.debug("cid({}): hit full page", cid);
                                HotListByTopicResponse response = new HotListByTopicResponse(
                                        0, "成功");
                                response.setList(fetch_page_result.getPage());
                                return response;
                            }
                            logger.debug("cid({}): not enough page", cid);
                        }
                    }
                    reads = filter_topic_hots_reads(cid_node.getRead_xids(),
                            topics);
                }
            }
            else
            {
                need_load = true;
                logger.debug("cid({}): need_load", cid);
            }
        }
        finally
        {
            cid_hash.getLock().readLock().unlock();
        }

        // 载入cid
        Map<String, Instant> read_xids_load = new HashMap<>();
        Map<String, Instant> dislike_xids_load = new HashMap<>();
        if (need_load)
        {
            LoadCidResult load_cid_result = load_cid(cid);
            if (load_cid_result != null)
            {
                read_xids_load = load_cid_result.getRead_xids();
                dislike_xids_load = load_cid_result.getDislike_xids();
                reads = filter_topic_hots_reads(read_xids_load, topics);
                logger.debug("cid({}): load, read({}), dislike({})", cid,
                        read_xids_load.size(), dislike_xids_load.size());
            }
        }

        // 同步xids
        SyncTopicHotsResult sync_topic_hots_result = sync_topic_hots(cid,
                topics, device, agent, reads);

        try
        {
            cid_hash.getLock().writeLock().lock();

            long modify_begin_time = Instant.now().toEpochMilli();

            CidNode cid_node = cid_hash.getCid_nodes().get(cid);
            if (cid_node == null)
            {
                cid_node = new CidNode();
                cid_node.setRead_xids(read_xids_load);
                cid_node.setDislike_xids(dislike_xids_load);
                cid_hash.getCid_nodes().put(cid, cid_node);
                logger.debug("cid({}): add to hash", cid);
            }
            cid_node.setTopic_hots_sync_time(now);
            Map<Integer, Map<String, Integer>> topic_hots = cid_node
                    .getTopic_hots();
            topic_hots.putAll(sync_topic_hots_result.getHots());

            HotListByTopicResponse response = new HotListByTopicResponse(0,
                    "成功");
            FetchPageResult fetch_page_result = fetch_topic_hot_page(cid_node,
                    topics);
            List<HotResponseNode> page = fetch_page_result.getPage();
            if (!page.isEmpty())
            {
                // 修改已读列表
                modify_read(fetch_page_result.getPage(), now, 0,
                        cid_node.getRead_xids(), cid_node.getRead_log());
                // cid_node.setAccess_time(now);
                response.setList(page);
                logger.debug("cid({}): sync page", cid);
            }
            else
            {
                if (is_first == 1)
                { // 首次进入无记录则显示最后一页
                    modify_read(fetch_page_result.getPage(), now, 1,
                            cid_node.getRead_xids(), cid_node.getRead_log());
                    response.setList(fetch_last_topic_hot_page(topics,
                            cid_node.getRead_xids()));
                }
                else
                {
                    response.setList(new ArrayList<>());
                }
                logger.debug("cid({}): sync empty page, is_first({})", cid,
                        is_first);
            }

            int cost = (int) (Instant.now().toEpochMilli() - modify_begin_time);
            if (cost > 50)
            {
                logger.info("cid({}): modify cid_node cost {} ms", cid, cost);
            }

            return response;
        }
        finally
        {
            cid_hash.getLock().writeLock().unlock();
        }
    }

    private FetchPageResult fetch_topic_hot_page(CidNode cid_node,
            List<Integer> topics)
    {
        Map<Integer, Map<String, Integer>> hots = cid_node.getTopic_hots();
        List<String> hots_topics = topics.stream()
                .flatMap(topic -> Optional.ofNullable(hots.get(topic))
                        // .orElse(Optional.ofNullable(topic_hots_default.get(topic))
                        // .orElse(new HashMap<>()))
                        .orElse(new HashMap<>()).entrySet().stream())
                .sorted((x, y) -> -x.getValue().compareTo(y.getValue()))
                .map(Entry::getKey).collect(Collectors.toList());

        try
        {
            data_lock.readLock().lock();

            FetchPageResult result = new FetchPageResult();

            Map<String, Instant> read_xids = cid_node.getRead_xids();
            Map<String, Instant> dislike_xids = cid_node.getDislike_xids();
            List<HotResponseNode> page = hots_topics.stream().map(xid -> {
                if (read_xids != null && read_xids.containsKey(xid))
                    return null;
                if (dislike_xids != null && dislike_xids.containsKey(xid))
                    return null;
                if (blacks.containsKey(xid))
                    return null;
                ContentNode cn = contents.get(xid);
                if (cn == null)
                {
                    return null;
                }
                HotResponseNode node = new HotResponseNode();
                node.setXid(xid);
                node.setContent(
                        Optional.ofNullable(cn.getContent()).orElse("{}"));
                return node;
            }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            result.setPage(page);
            return result;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private List<HotResponseNode> fetch_last_topic_hot_page(
            List<Integer> topics, Map<String, Instant> read_xids)
    {
        try
        {
            data_lock.readLock().lock();

            List<HotResponseNode> page = read_xids.entrySet().stream()
                    .sorted((x, y) -> -x.getValue().compareTo(y.getValue()))
                    .map(entry -> {
                        String xid = entry.getKey();
                        ContentNode cn = contents.get(xid);
                        if (cn == null)
                            return null;
                        if (!topics.isEmpty() && !topics.contains(0)
                                && !topics.contains(cn.getTopic()))
                            return null;
                        HotResponseNode node = new HotResponseNode();
                        node.setXid(xid);
                        node.setContent(Optional.ofNullable(cn.getContent())
                                .orElse("{}"));
                        return node;
                    }).filter(Objects::nonNull).limit(PAGE_SIZE)
                    .collect(Collectors.toList());
            Collections.reverse(page);
            return page;
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    private List<String> filter_topic_hots_reads(Map<String, Instant> read_xids,
            List<Integer> topics)
    {
        if (topics == null || topics.isEmpty() || topics.contains(0))
        {
            return read_xids.keySet().stream().collect(Collectors.toList());
        }

        try
        {
            data_lock.readLock().lock();

            return read_xids.keySet().stream().filter(xid -> {
                ContentNode cn = contents.get(xid);
                if (cn == null)
                    return false;
                if (topics.contains(cn.getTopic()))
                    return false;
                return true;
            }).collect(Collectors.toList());
        }
        finally
        {
            data_lock.readLock().unlock();
        }
    }

    class SyncTopicHotsResult
    {
        private Map<Integer, Map<String, Integer>> hots = new HashMap<>();
        private long size;

        public long getSize()
        {
            return size;
        }

        public void setSize(long size)
        {
            this.size = size;
        }

        public Map<Integer, Map<String, Integer>> getHots()
        {
            return hots;
        }

        public void setHots(Map<Integer, Map<String, Integer>> hots)
        {
            this.hots = hots;
        }

    }

    private SyncTopicHotsResult sync_topic_hots(int cid, List<Integer> topics,
            String device, String agent, List<String> reads)
    {
        try
        {
            long http_begin_time = Instant.now().toEpochMilli();

            SyncTopicHotsRequest request = new SyncTopicHotsRequest();
            request.setCid(cid);
            if (topics != null && !topics.isEmpty() && !topics.contains(0))
            {
                request.setTopics(topics);
            }
            if (reads != null && !reads.isEmpty())
            {
                request.setRead_xids(reads);
            }
            request.setDevice(Optional.ofNullable(device).orElse(""));
            request.setAgent(Optional.ofNullable(agent).orElse(""));
            String response_s = post(appconfig.http.hotlistbytopic,
                    objectMapper.writeValueAsString(request));
            SyncTopicHotsResponse response = objectMapper.readValue(response_s,
                    SyncTopicHotsResponse.class);
            SyncTopicHotsResult result = new SyncTopicHotsResult();
            result.setSize(response_s.getBytes().length);
            result.setHots(response.getData());

            int cost = (int) (Instant.now().toEpochMilli() - http_begin_time);
            if (cost > 100)
            {
                logger.info(
                        "cid({}) topics({}): sync topichots http cost {} ms,"
                                + " reads({}), num({}) size({})",
                        cid,
                        Optional.ofNullable(topics).orElse(new ArrayList<>())
                                .stream().map(String::valueOf)
                                .collect(Collectors.joining(",")),
                        cost,
                        Optional.ofNullable(reads).orElse(new ArrayList<>())
                                .size(),
                        result.getHots().values().stream()
                                .collect(Collectors.summingInt(Map::size)),
                        result.getSize());
            }

            return result;
        }
        catch (Exception e)
        {
            logger.info("cid({}) sync fail: {}", cid, e);
            return new SyncTopicHotsResult();
        }
    }
}

class SyncHotListRequest
{
    private int cid;
    private List<Integer> fids = new ArrayList<>();
    private List<String> read_xids = new ArrayList<>();
    private String device;
    private String agent;

    public int getCid()
    {
        return cid;
    }

    public void setCid(int cid)
    {
        this.cid = cid;
    }

    public List<Integer> getFids()
    {
        return fids;
    }

    public void setFids(List<Integer> fids)
    {
        this.fids = fids;
    }

    public List<String> getRead_xids()
    {
        return read_xids;
    }

    public void setRead_xids(List<String> read_xids)
    {
        this.read_xids = read_xids;
    }

    public String getDevice()
    {
        return device;
    }

    public void setDevice(String device)
    {
        this.device = device;
    }

    public String getAgent()
    {
        return agent;
    }

    public void setAgent(String agent)
    {
        this.agent = agent;
    }

}

class SyncHotListResponse
{
    // { fid: { xid: score } }
    private Map<Integer, Map<String, Integer>> data;

    public Map<Integer, Map<String, Integer>> getData()
    {
        return data;
    }

    public void setData(Map<Integer, Map<String, Integer>> data)
    {
        this.data = data;
    }

}

class SyncVideoListRequest
{
    private int cid;
    private List<String> read_xids = new ArrayList<>();

    public int getCid()
    {
        return cid;
    }

    public void setCid(int cid)
    {
        this.cid = cid;
    }

    public List<String> getRead_xids()
    {
        return read_xids;
    }

    public void setRead_xids(List<String> read_xids)
    {
        this.read_xids = read_xids;
    }

}

class SyncVideoListResponse
{
    // { xid: score }
    private Map<String, Integer> data;

    public Map<String, Integer> getData()
    {
        return data;
    }

    public void setData(Map<String, Integer> data)
    {
        this.data = data;
    }

}

class SyncTopicHotsRequest
{
    private int cid;
    private List<Integer> topics = new ArrayList<>();
    private List<String> read_xids = new ArrayList<>();
    private String device;
    private String agent;

    public int getCid()
    {
        return cid;
    }

    public void setCid(int cid)
    {
        this.cid = cid;
    }

    public List<Integer> getTopics()
    {
        return topics;
    }

    public void setTopics(List<Integer> topics)
    {
        this.topics = topics;
    }

    public List<String> getRead_xids()
    {
        return read_xids;
    }

    public void setRead_xids(List<String> read_xids)
    {
        this.read_xids = read_xids;
    }

    public String getDevice()
    {
        return device;
    }

    public void setDevice(String device)
    {
        this.device = device;
    }

    public String getAgent()
    {
        return agent;
    }

    public void setAgent(String agent)
    {
        this.agent = agent;
    }

}

class SyncTopicHotsResponse
{
    // { topic: { xid: score } }
    private Map<Integer, Map<String, Integer>> data;

    public Map<Integer, Map<String, Integer>> getData()
    {
        return data;
    }

    public void setData(Map<Integer, Map<String, Integer>> data)
    {
        this.data = data;
    }

}

class HotResponseNode
{
    private String xid;
    @JsonRawValue
    private String content = "";

    public String getXid()
    {
        return xid;
    }

    public void setXid(String xid)
    {
        this.xid = xid;
    }

    public String getContent()
    {
        return content;
    }

    public void setContent(String content)
    {
        this.content = content;
    }

}

class HotListResponse extends BaseResponse
{
    public HotListResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<HotResponseNode> list = new ArrayList<>();

    public List<HotResponseNode> getList()
    {
        return list;
    }

    public void setList(List<HotResponseNode> list)
    {
        this.list = list;
    }

}

class DislikeResponse extends BaseResponse
{
    public DislikeResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

}

class ClearReadResponse extends BaseResponse
{
    public ClearReadResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

}

class GetReadResponse extends BaseResponse
{
    public GetReadResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<GetReadResponseNode> list = new ArrayList<>();

    public List<GetReadResponseNode> getList()
    {
        return list;
    }

    public void setList(List<GetReadResponseNode> list)
    {
        this.list = list;
    }

}

class GetReadResponseNode
{
    private String xid;
    private String read_time;

    public String getXid()
    {
        return xid;
    }

    public void setXid(String xid)
    {
        this.xid = xid;
    }

    public String getRead_time()
    {
        return read_time;
    }

    public void setRead_time(String read_time)
    {
        this.read_time = read_time;
    }

}

class GetDislikeResponse extends BaseResponse
{
    public GetDislikeResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<String> list = new ArrayList<>();

    public List<String> getList()
    {
        return list;
    }

    public void setList(List<String> list)
    {
        this.list = list;
    }

}

class GetListResponse extends BaseResponse
{
    public GetListResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<GetListResponseNode> list = new ArrayList<>();

    public List<GetListResponseNode> getList()
    {
        return list;
    }

    public void setList(List<GetListResponseNode> list)
    {
        this.list = list;
    }

}

class GetListResponseNode
{
    public final static class Status
    {
        public final static int READ = 1; // 已读
        public final static int DISLIKE = 2; // 不喜欢
        public final static int NO_CONTENT = 3; // 无内容
        public final static int BLACK = 4; // 黑名单
    }

    private String xid;
    private int status;
    private int fid;

    public String getXid()
    {
        return xid;
    }

    public void setXid(String xid)
    {
        this.xid = xid;
    }

    public int getStatus()
    {
        return status;
    }

    public void setStatus(int status)
    {
        this.status = status;
    }

    public int getFid()
    {
        return fid;
    }

    public void setFid(int fid)
    {
        this.fid = fid;
    }

}

// class GetStatResponse extends BaseResponse
// {
// public GetStatResponse(int errno, String errmsg)
// {
// super(errno, errmsg);
// }
//
// }

class VideoListResponse extends BaseResponse
{
    public VideoListResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<HotResponseNode> list = new ArrayList<>();

    public List<HotResponseNode> getList()
    {
        return list;
    }

    public void setList(List<HotResponseNode> list)
    {
        this.list = list;
    }

}

class RelatedVideoListResponse extends BaseResponse
{
    public RelatedVideoListResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private HotResponseNode curr;
    private List<HotResponseNode> list = new ArrayList<>();

    public HotResponseNode getCurr()
    {
        return curr;
    }

    public void setCurr(HotResponseNode curr)
    {
        this.curr = curr;
    }

    public List<HotResponseNode> getList()
    {
        return list;
    }

    public void setList(List<HotResponseNode> list)
    {
        this.list = list;
    }

}

class UnreadResponse extends BaseResponse
{

    public UnreadResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

}

class HotListByTopicResponse extends BaseResponse
{
    public HotListByTopicResponse(int errno, String errmsg)
    {
        super(errno, errmsg);
    }

    private List<HotResponseNode> list = new ArrayList<>();

    public List<HotResponseNode> getList()
    {
        return list;
    }

    public void setList(List<HotResponseNode> list)
    {
        this.list = list;
    }

}
